package com.john.elasticsearch.common;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.parser.Feature;
import com.john.utils.page.PageInfoUtils;
import com.john.utils.token.TokenSing;
import org.apache.http.HttpHost;
import org.elasticsearch.action.ActionListener;
import org.elasticsearch.action.admin.indices.delete.DeleteIndexRequest;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.delete.DeleteRequest;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.search.*;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.client.core.CountRequest;
import org.elasticsearch.client.core.CountResponse;
import org.elasticsearch.client.indices.CreateIndexRequest;
import org.elasticsearch.client.indices.GetIndexRequest;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.common.xcontent.XContentType;
import org.elasticsearch.index.query.BoolQueryBuilder;
import org.elasticsearch.search.Scroll;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.elasticsearch.search.sort.SortOrder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.data.domain.Example;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.TimeUnit;

/**
 * ES客户端
 */
public class ESClients {
    public static Logger logger = LoggerFactory.getLogger(ESClients.class);
    private static ThreadLocal<RestHighLevelClient> resthighlevelclientThreadLocal = new ThreadLocal<RestHighLevelClient>();

    /**
     * 获取RestHighLevelClient
     * @return
     */
    public static RestHighLevelClient getClientConnection() {
        RestHighLevelClient restHighLevelClient = resthighlevelclientThreadLocal.get();
        try {
            if(null == restHighLevelClient){
                restHighLevelClient  = new RestHighLevelClient(RestClient.builder(
                        new HttpHost("40.26.204.6",9200)
                ));
                resthighlevelclientThreadLocal.set(restHighLevelClient);
            }
        } catch (Exception e) {
            logger.error("[RestHighLevelClient Exception] --> "
                    + "Failed to create a RestHighLevelClient, the exceprion message is:" + e.getMessage());
        }
        return restHighLevelClient;
    }

    /**
     * 关闭restHighLevelClient
     */
    public static void closeRestHighLevelClient() {
        RestHighLevelClient restHighLevelClient = resthighlevelclientThreadLocal.get();
        if(null != restHighLevelClient){
            try {
                restHighLevelClient.close();
                resthighlevelclientThreadLocal.remove();
            } catch (Exception e) {
                logger.error("[RestHighLevelClient Exception] --> "
                        + "Failed to close the RestHighLevelClient, the exceprion message is:" + e.getMessage());
            }
        }
    }

    /**
     * 创建索引
     * @param indexName 索引名称
     * @return
     */
    public static boolean createIndex(String indexName) {
        try {
            indexName = indexName.toLowerCase();
            if(!existIndex(indexName)){
                CreateIndexRequest createIndexRequest = new CreateIndexRequest(indexName);
                //设置分片和副本数
                createIndexRequest.settings(Settings.builder()
                        .put("index.number_of_shards", 5)
                        .put("index.number_of_replicas", 1)
                );

                getClientConnection().indices().create(createIndexRequest, RequestOptions.DEFAULT);
                return true;
            }
        }catch (Exception ex){
            ex.printStackTrace();
            return  false;
        }finally {
            closeRestHighLevelClient();
        }
        return false;
    }

    /**
     * 查询索引是否存在
     * @param indexName 索引名称
     * @return
     */
    public static boolean existIndex(String indexName) {
        try {
            indexName = indexName.toLowerCase();
            GetIndexRequest request = new GetIndexRequest(indexName);
            return getClientConnection().indices().exists(request, RequestOptions.DEFAULT);
        }catch (Exception ex){
            ex.printStackTrace();
            return false;
        }finally {
            closeRestHighLevelClient();
        }
    }

    /**
     * 删除索引
     * @param indexName 索引名称
     * @return
     */
    public static boolean removeIndex(String indexName) {
        try {
            indexName = indexName.toLowerCase();
            DeleteIndexRequest deleteIndexRequest = new DeleteIndexRequest(indexName);
            getClientConnection().indices().delete(deleteIndexRequest, RequestOptions.DEFAULT);
        }catch (Exception e){
            e.printStackTrace();
            return false;
        }finally {
            closeRestHighLevelClient();
        }
        return true;
    }

    /**
     * 同步数据到ES搜索引擎
     * @param indexName
     * @param source
     * @param pageInfoUtils
     */
    public static void insertDocumentTOES(String indexName, JSONArray source, PageInfoUtils pageInfoUtils){
        try {
            long start = System.currentTimeMillis();
            indexName = indexName.toLowerCase();
            System.out.println("线程" + Thread.currentThread().getId() + "开始执行ES搜索引擎数据批量添加");
            BulkRequest bulkRequest = new BulkRequest();
            bulkRequest.timeout("2m");
            //刷新策略
            bulkRequest.setRefreshPolicy("wait_for");//字符串方式

            for (int i = 0; i < source.size(); i++) {
                String id = String.valueOf(pageInfoUtils.getStartRow() + i + 1);
                bulkRequest.add( new IndexRequest(indexName).id(id).source(source.getJSONObject(i).toJSONString(),XContentType.JSON));
            }
            getClientConnection().bulkAsync(bulkRequest,RequestOptions.DEFAULT,new ActionListener<BulkResponse>() {
                @Override
                public void onResponse(BulkResponse bulkResponse) {
                    //执行成功完成时调用。 response作为参数提供，并包含已执行的每个操作的单个结果列表。 请注意，一个或多个操作可能已失败，然而其他操作已成功执行。
                    System.out.println("异步同步数据到ES搜索引擎成功...");
                }

                @Override
                public void onFailure(Exception e) {
                    //在整个BulkRequest失败时调用。 在这种情况下，exception作为参数提供，并且没有执行任何操作。
                    e.printStackTrace();
                    System.out.println("异步同步数据到ES搜索引擎失败...");
                }
            });
            System.out.println("线程" + Thread.currentThread().getId() + "执行ES搜索引擎数据批量添加结束,共计添加了"+source.size()+"条数据,当前执行分页明细:"+JSON.toJSONString(pageInfoUtils));
            long end = System.currentTimeMillis();
            System.out.println("耗时:"+(end - start) / (1000 * 60)+"分钟");
        }catch (Exception ex){
            ex.printStackTrace();
        }finally {
            closeRestHighLevelClient();
        }
    }



    public static List<String> queryPage(String indexName,Object object,PageInfoUtils pageInfoUtils) {
        List<String> rows = new ArrayList<>();
        try {
            CountRequest countRequest = new CountRequest(indexName);
            BoolQueryBuilder boolQueryBuilder = new BoolQueryBuilder();
            countRequest.query(boolQueryBuilder);
            CountResponse countResponse = getClientConnection().count(countRequest,RequestOptions.DEFAULT);
            pageInfoUtils.setTotalRecord(Long.valueOf(countResponse.getCount()).intValue());
            if(pageInfoUtils.validator()){
                SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
                searchSourceBuilder.query(boolQueryBuilder);
                //起始位置
                searchSourceBuilder.from(pageInfoUtils.getStartRow());
                //每页数量
                searchSourceBuilder.size(pageInfoUtils.getPageSize());
                SearchRequest request = new SearchRequest();
                request.indices(indexName);
                request.source(searchSourceBuilder);
                SearchResponse response =  getClientConnection().search(request, RequestOptions.DEFAULT);
                SearchHit[] hits = response.getHits().getHits();
                for (SearchHit hit : hits) {
                    rows.add(hit.getSourceAsString());
                }
//                pageInfoUtils.setRows(rows);
            }
        }catch (Exception ex){
            ex.printStackTrace();
        }finally {
            closeRestHighLevelClient();
        }
        return rows;
    }

    public static void testScroll(String indexName, Object t, PageInfoUtils pageInfoUtils) {
        try {
            RestHighLevelClient client = getClientConnection();
            CountRequest countRequest = new CountRequest(indexName);
            SearchSourceBuilder searchSourceBuilder = null;
            BoolQueryBuilder boolQueryBuilder = Utils.getBoolQueryBuilder(t);
            countRequest.query(boolQueryBuilder);
            CountResponse countResponse = client.count(countRequest, RequestOptions.DEFAULT);
            pageInfoUtils.setTotalRecord(Long.valueOf(countResponse.getCount()).intValue());
            if(pageInfoUtils.validator()) {
               if (searchSourceBuilder == null) {
                    searchSourceBuilder = new SearchSourceBuilder();
                }
                // 初始化scroll
                // 设定滚动时间间隔
                // 这个时间并不需要长到可以处理所有的数据，仅仅需要足够长来处理前一批次的结果。每个 scroll 请求（包含 scroll 参数）设置了一个新的失效时间。
                final Scroll scroll = new Scroll(TimeValue.timeValueMinutes(10L));
                SearchRequest searchRequest = new SearchRequest(indexName); // 新建索引搜索请求
                searchRequest.scroll(scroll);
                searchSourceBuilder.query(boolQueryBuilder);
                searchSourceBuilder.size(pageInfoUtils.getPageSize()); //设定每次返回多少条数据
//              searchSourceBuilder.fetchSource(new String[]{"nid"},null);//设置返回字段和排除字段
                searchRequest.source(searchSourceBuilder);
                SearchResponse searchResponse = client.search(searchRequest, RequestOptions.DEFAULT);
                //初始页数
                int page = 1;
                SearchHit[] searchHits = searchResponse.getHits().getHits();
                //遍历搜索命中的数据，直到没有数据
                String scrollId = searchResponse.getScrollId();
                while (searchHits != null && searchHits.length > 0) {
                    SearchScrollRequest scrollRequest = new SearchScrollRequest(scrollId);
                    scrollRequest.scroll(scroll);
                    searchResponse = client.scroll(scrollRequest, RequestOptions.DEFAULT);
                    scrollId = searchResponse.getScrollId();
                    searchHits = searchResponse.getHits().getHits();
                    if (searchHits != null && searchHits.length > 0) {
                        page++;
                        System.out.println("-----第" + page + "页-----scrollId:------"+scrollId);
                        for (SearchHit searchHit : searchHits) {
                            String sourceAsString = searchHit.getSourceAsString();
//                            System.out.println(sourceAsString);
                            JSONObject jsonObject = JSONObject.parseObject(sourceAsString);
                            System.out.println(jsonObject.get("uniqueid"));
                        }
                    }
                }
                //清除滚屏
                ClearScrollRequest clearScrollRequest = new ClearScrollRequest();
                clearScrollRequest.addScrollId(scrollId);//也可以选择setScrollIds()将多个scrollId一起使用
                ClearScrollResponse clearScrollResponse = null;
                clearScrollResponse = client.clearScroll(clearScrollRequest, RequestOptions.DEFAULT);
                boolean succeeded = clearScrollResponse.isSucceeded();
                System.out.println("succeeded:" + succeeded);
                System.out.println("总数量:"+Long.valueOf(countResponse.getCount()).intValue());
            }
        } catch (Exception e) {
            e.printStackTrace();
        }finally {
            closeRestHighLevelClient();
        }
    }

    public static JSONArray testScrolls(String indexName, Object t, PageInfoUtils pageInfoUtils) {
        JSONArray jsonArray = new JSONArray();
        try {
            RestHighLevelClient client = getClientConnection();
            CountRequest countRequest = new CountRequest(indexName);
            BoolQueryBuilder boolQueryBuilder = Utils.getBoolQueryBuilder(t);
            countRequest.query(boolQueryBuilder);
            CountResponse countResponse = client.count(countRequest, RequestOptions.DEFAULT);
            pageInfoUtils.setTotalRecord(Long.valueOf(countResponse.getCount()).intValue());
            if(pageInfoUtils.validator()) {
                SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
                // 初始化scroll
                // 设定滚动时间间隔
                // 这个时间并不需要长到可以处理所有的数据，仅仅需要足够长来处理前一批次的结果。每个 scroll 请求（包含 scroll 参数）设置了一个新的失效时间。
                final Scroll scroll = new Scroll(TimeValue.timeValueMinutes(10L));
                SearchRequest searchRequest = new SearchRequest(indexName); // 新建索引搜索请求
                searchRequest.scroll(scroll);
                searchSourceBuilder.query(boolQueryBuilder);
                searchSourceBuilder.size(pageInfoUtils.getPageSize()); //设定每次返回多少条数据
                // 设置请求超时时间
                searchSourceBuilder.timeout(new TimeValue(20, TimeUnit.SECONDS));
                searchRequest.source(searchSourceBuilder);
                SearchResponse searchResponse = client.search(searchRequest, RequestOptions.DEFAULT);
                String scrollId = searchResponse.getScrollId();
                for (int i = 1; i <= pageInfoUtils.getTotalPage(); i++) {
                    System.out.println("scrollId:"+scrollId +"========当前页数:"+i);
                    /*if(i == pageInfoUtils.getPageNum()){ //当前页
                        //获取结果
                        SearchHit[] hits = searchResponse.getHits().getHits();
                        JSONObject jsonObject = null;
                        for (SearchHit hit : hits) {
                            jsonObject = JSON.parseObject(hit.getSourceAsString(),Feature.OrderedField);
                            jsonArray.add(jsonObject);
                        }
                        break;
                    }*/
                    SearchScrollRequest scrollRequest = new SearchScrollRequest(scrollId);
                    scrollRequest.scroll(scroll);
                    searchResponse = client.scroll(scrollRequest, RequestOptions.DEFAULT);
                    scrollId = searchResponse.getScrollId();
                }
                //清除快照
                ClearScrollRequest clearScrollRequest = new ClearScrollRequest();
                clearScrollRequest.addScrollId(scrollId);//也可以选择setScrollIds()将多个scrollId一起使用
                ClearScrollResponse clearScrollResponse = null;
                clearScrollResponse = client.clearScroll(clearScrollRequest, RequestOptions.DEFAULT);
                boolean succeeded = clearScrollResponse.isSucceeded();
                System.out.println("succeeded:" + succeeded);
            }
        } catch (Exception e) {
            e.printStackTrace();
        }finally {
            closeRestHighLevelClient();
        }
        return jsonArray;
    }


    /**searchAfter*/
    public static JSONArray searchAfter(String indexName, Object t, Example example,SearchAfterResult searchAfterResult){
        JSONArray jsonArray = new JSONArray();
        try {
            if(searchAfterResult.validator()){
                Object[] objects = null;
                BoolQueryBuilder boolQueryBuilder = Utils.getBoolQueryBuilder(t);
                SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
                searchSourceBuilder.query(boolQueryBuilder);
                searchSourceBuilder.size(10);
                searchSourceBuilder.sort("_id", SortOrder.ASC);
                //如果交替值不为空则按交替值分页查询，反之只查第一页
                if(searchAfterResult.getHands() != null && !searchAfterResult.getHands().equals("")) {
                    objects = new Object[]{TokenSing.decrypt(searchAfterResult.getHands())};
                    searchSourceBuilder.searchAfter(objects);
                }
                SearchRequest searchRequest = new SearchRequest(indexName);
                searchRequest.source(searchSourceBuilder);
                SearchResponse searchResponse = getClientConnection().search(searchRequest, RequestOptions.DEFAULT);
                SearchHit[] hits = searchResponse.getHits().getHits();
                if(hits.length > 0){
                    searchAfterResult.setHands(String.valueOf(hits[hits.length-1].getSortValues()[0]));
                    JSONObject jsonObject = null;
                    for (SearchHit searchHit : hits) {
                        jsonObject = JSON.parseObject(searchHit.getSourceAsString(),Feature.OrderedField);
                        System.out.println(jsonObject.get("indexDataId") + "===============" + String.valueOf(hits[hits.length-1].getSortValues()[0]));
                        jsonArray.add(jsonObject);
                    }
                }else{
                    searchAfterResult.setHands(searchAfterResult.getHandsDefault());
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        }finally {
            closeRestHighLevelClient();
        }
        return jsonArray;
    }

    /**
     * 根据索引名称 数据编号删除数据
     * @param indexName
     * @param indexId
     */
    public static void removeById(String indexName, String indexId) {
        try {
            DeleteRequest request = new DeleteRequest(indexName, indexId);
            getClientConnection().delete(request, RequestOptions.DEFAULT);
        }catch (Exception ex){
            logger.error("根据索引数据ID删除数据失败");
            logger.error(ex.getMessage(),ex);
            ex.printStackTrace();
        }finally {
            closeRestHighLevelClient();
        }
    }


    public static void main(String[] args) {
//        createIndex("di_adi_rec_docindex_hr");
        SearchAfterResult searchAfterResult = new SearchAfterResult();
        searchAfterResult.setHands("9999");
        JSONArray jsonArray = searchAfter("di_adi_rec_docindex_hr",null,null,searchAfterResult);
        if(searchAfterResult.validator()){
//            List<Di_Adi_Rec_Docindex_Hr> list = JSONObject.parseArray(jsonArray.toJSONString(), Di_Adi_Rec_Docindex_Hr.class);
//            searchAfterResult.setRows(list);
//            System.out.println(TokenSing.decrypt("6c73d284b1680a9097bc77e3951e28275246abd4da421e30da33bdb8c4097a78"));
//            System.out.println(JSON.toJSONString(new AjaxResult(com.medical.common.enums.ResultStatus.SUCCESS.getCode(), com.medical.common.enums.ResultStatus.SUCCESS.getMsg(),searchAfterResult)));
        }else{
            System.out.println(JSON.toJSONString(searchAfterResult.getAjaxResult()));
        }
    }

}
